package com.mpgame.mq;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;

import com.google.protobuf.InvalidProtocolBufferException;
import com.mpgame.common.mq.KafkaCfg;
import com.mpgame.common.mq.KafkaTopics;
import com.mpgame.common.proto.MQPacket;
import com.mpgame.common.proto.PlayerLeaveMsg;
import com.mpgame.common.utils.Log;
import com.mpgame.service.BattleSvc;

public class MessageQueueConsumer {
    private static volatile MessageQueueConsumer instance;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private volatile boolean isRunning = false;

    private KafkaConsumer<String, MQPacket> kConsumer;


    private MessageQueueConsumer(){}

    public static MessageQueueConsumer getInstance(){
        if(instance == null){
            synchronized(MessageQueueConsumer.class){
                if(instance == null){
                    instance = new MessageQueueConsumer();
                }
            }
        }
        return instance;
    }

    public void start(){
        if(isRunning){
            Log.warn("消息消费者已经在运行中");
            return;
        }
        isRunning = true;
        Properties p1 = KafkaCfg.baseConsumerProps("mp-battle");
        kConsumer = new KafkaConsumer<>(p1);
        kConsumer.subscribe(Collections.singletonList(KafkaTopics.NOTIFY_TO_SERVER));
        scheduler.scheduleWithFixedDelay(this::pollMqPacket, 0, 20, TimeUnit.MILLISECONDS);
        Log.info("消息队列消费者已启动");
    }

    private void pollMqPacket(){
        try{
            ConsumerRecords<String, MQPacket> records = kConsumer.poll(Duration.ofMillis(100));
            for(ConsumerRecord<String, MQPacket> rec : records){
                MQPacket pkt = rec.value();
                handleMqPacket(pkt);
            }
            kConsumer.commitAsync();
        }catch(WakeupException ignored){
        }catch(Exception e){
            Log.err("pollMqPacket异常：" + e.getMessage());
        }
    }
    private void handleMqPacket(MQPacket pkt){
        Log.info("开始处理消息队列包：" + pkt);
        switch(pkt.getCmd()){
            case GAME_PLAYER_LEAVE:
                handleGamePlayerLeave(pkt);
                break;
            default:
                Log.warn("Battle消息队列消费者不处理队列包：" + pkt);
                break;
        }
    }
    private void handleGamePlayerLeave(MQPacket pkt){
        try{
            PlayerLeaveMsg msg = PlayerLeaveMsg.parseFrom(pkt.getData());
            BattleSvc.GetInstance().handlePlayerOffline(msg.getAccountId());
        }catch(InvalidProtocolBufferException e){Log.err("消息解析失败！");}
    }
    public void stop(){
        if(!isRunning) return;
        isRunning = false;
        try{ if(kConsumer != null) kConsumer.wakeup();}catch(Exception ignore){}
        scheduler.shutdown();
        Log.info("消息队列消费者已停止！");
    }
}
